
package com.etlmaster.executor.utils;

import java.util.List;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

import com.etlmaster.executor.utils.LogWriter;

/**
 * @description zk连接处理实现类(Curator实现)
 */
public class ZKUtil
{
private static ZKUtil etlZk;
	//public static final int DEFAULT_SESSION_TIMEOUT = 5000;// zooKeeper session的超时时间
	public static final int DEFAULT_SESSION_TIMEOUT = DataCollectionUtils.getIntProperties("zktimeout");// zooKeeper session的超时时间
	public static final int DEFAULT_CONN_TIMEOUT = 30000;// 连接zooKeeper的超时时间

	public static final int create_ok=0;
	public static final int create_failed=1;
	private CuratorFramework client;

	private ZKUtil(String zkConnStr)
	{
		
		if (client == null)
		{
			try
			{
				client = CuratorFrameworkFactory.newClient(zkConnStr, DEFAULT_SESSION_TIMEOUT, DEFAULT_CONN_TIMEOUT, new RetryNTimes(2000, 20000));
				client.start();
			}
			catch (Exception e)
			{
				LogWriter.addLog("Error","InterruptedException happend!!!{}", e.getMessage());
			}

		}
	}

	public synchronized static ZKUtil getInstance()
	{
		if (etlZk == null)
		{
			etlZk = new ZKUtil(null);
		}

		return etlZk;
	}

	public synchronized static ZKUtil getInstance(String zkConnectStr)
	{
		if (etlZk == null)
		{
			etlZk = new ZKUtil(zkConnectStr);
		}

		return etlZk;
	}


	public void createP(String znode, String zdata)
	{
		create(znode, zdata, CreateMode.PERSISTENT);
	}

	public int createE(String znode, String zdata)
	{
		return create(znode, zdata, CreateMode.EPHEMERAL);
	}

	public int create(String znode, String zdata, CreateMode mode)
	{

		try
		{
			if (null != zdata)
			{
				client.create().creatingParentsIfNeeded().withMode(mode).withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).forPath(znode, zdata.getBytes());
			}
			else
			{
				client.create().creatingParentsIfNeeded().withMode(mode).withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).forPath(znode);
			}
			return create_ok;
		}
		catch (Exception e)
		{
			LogWriter.addLog("ERROR","创建[" + znode + "]节点失败！{}", e.getMessage());
			return create_failed;
		}
	}

	public Stat exists(String znode)
	{
		Stat stat = null;
		try
		{
			stat = client.checkExists().forPath(znode);
		}
		catch (Exception e)
		{
			LogWriter.addLog("ERROR","获取节点[" + znode + "]信息失败！{}", e.getMessage());
		}
		return stat;
	}

	public void delete(String znode)
	{
		try
		{
			client.delete().forPath(znode);
		}
		catch (Exception e)
		{
			LogWriter.addLog("ERROR","删除节点[" + znode + "]失败！{}", e.getMessage());
		}
	}

	public List<String> getChildren(String znode, Watcher watcher)
	{
		List<String> chailren = null;
		try
		{
			chailren = client.getChildren().usingWatcher(watcher).forPath(znode);
		}
		catch (Exception e)
		{
			LogWriter.addLog("ERROR","获取[" + znode + "]的子节点失败！{}", e.getMessage());
		}
		return chailren;
	}

	public List<String> getChildren(String znode)
	{
		List<String> chailren = null;
		try
		{
			chailren = client.getChildren().forPath(znode);
		}
		catch (Exception e)
		{
			LogWriter.addLog("ERROR","获取[" + znode + "]的子节点失败！", e.getMessage());
		}
		return chailren;
	}

	public String getData(String znode, Watcher watch, Stat stat)
	{
		byte[] bytes = null;
		try
		{
			bytes = client.getData().storingStatIn(stat).usingWatcher(watch).forPath(znode);
		}
		catch (Exception e)
		{
			LogWriter.addLog("ERROR","获取节点[" + znode + "]存储的数据失败！", e.getMessage());
		}
		if (null != bytes)
		{
			return new String(bytes);
		}
		return null;
	}

}
